package com.swak.rabbit.retry;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.springframework.beans.factory.DisposableBean;

import com.swak.rabbit.EventBus;
import com.swak.rabbit.message.Message;
import com.swak.rabbit.message.PendingConfirm;
import com.swak.reactivex.threads.SwakThreadFactory;
import com.swak.timer.HashedWheelTimer;
import com.swak.timer.Timeout;
import com.swak.timer.Timer;
import com.swak.timer.TimerTask;

/**
 * 自动发送重试
 * 
 * @author lifeng
 * @date 2021年3月3日 上午9:31:13
 */
public class AutoRetryPublisher implements RetryPublisher, DisposableBean {

	private final ConcurrentMap<String, RetryTask> failedPosted = new ConcurrentHashMap<>();
	// Timer for failure retry, regular check if there is a request for failure, and
	// if there is, an unlimited retry
	private final HashedWheelTimer retryTimer;
	private long retryPeriod = 5 * 1000;

	public AutoRetryPublisher() {
		// since the retry task will not be very much. 128 ticks is enough.
		retryTimer = new HashedWheelTimer(new SwakThreadFactory("RabbitMQ.RetryTimer-", true, new AtomicInteger(0)),
				retryPeriod, TimeUnit.MILLISECONDS, 128);
	}

	/**
	 * 有可能 failedPosted 中有数据，但是 retryTimer 已经重试了导致只能重试一次
	 */
	@Override
	public void add(PendingConfirm pendingConfirm) {
		RetryTask oldOne = failedPosted.get(pendingConfirm.getId());
		if (oldOne != null) {
			return;
		}
		RetryTask newTask = new RetryTask(pendingConfirm);
		oldOne = failedPosted.putIfAbsent(pendingConfirm.getId(), newTask);
		if (oldOne == null) {
			// never has a retry task. then start a new task for retry.
			retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
		}
	}

	@Override
	public void del(String id) {
		RetryTask f = failedPosted.remove(id);
		if (f != null) {
			f.cancel();
		}
	}

	public class RetryTask implements TimerTask {

		private final PendingConfirm pendingConfirm;
		private final int retryTimes;
		private volatile int times = 1;
		private volatile boolean cancel;

		public RetryTask(PendingConfirm pendingConfirm) {
			this.pendingConfirm = pendingConfirm;
			this.retryTimes = pendingConfirm.getRetryTimes();
			cancel = false;
		}

		public void cancel() {
			cancel = true;
		}

		public boolean isCancel() {
			return cancel;
		}

		@Override
		public void run(Timeout timeout) throws Exception {
			if (timeout.isCancelled() || timeout.timer().isStop() || isCancel()) {
				// other thread cancel this timeout or stop the timer.
				return;
			}
			if (times > retryTimes) {
				return;
			}
			try {
				doRetry(timeout).whenComplete((r, e) -> {
					if (e != null) {
						this.reput(timeout, retryPeriod);
					}
				});
			} catch (Throwable t) { // Ignore all the exceptions and wait for the next retry
				reput(timeout, retryPeriod);
			}
		}

		protected void reput(Timeout timeout, long tick) {
			if (timeout == null) {
				throw new IllegalArgumentException();
			}

			Timer timer = timeout.timer();
			if (timer.isStop() || timeout.isCancelled() || isCancel()) {
				return;
			}
			times++;
			timer.newTimeout(timeout.task(), tick, TimeUnit.MILLISECONDS);
		}

		/**
		 * 只保证在发送的过程中抛出异常则会重试, 再次发送的消息不参与重试
		 * 
		 * @param timeout
		 */
		protected CompletableFuture<Void> doRetry(Timeout timeout) {

			// 删除failedPosted中存储的超时任务
			failedPosted.remove(pendingConfirm.getId());

			// 发送数据
			return EventBus.me().submit(pendingConfirm.getExchange(), pendingConfirm.getRoutingKey(),
					Message.of().setId(pendingConfirm.getId()).setDeliveryMode(pendingConfirm.getDeliveryMode())
							.setExpiration(pendingConfirm.getExpiration()).setPriority(pendingConfirm.getPriority())
							.setPayload(pendingConfirm.getPayload()).setPublishRetrys(-1).build());
		}
	}

	@Override
	public void destroy() throws Exception {
		try {
			this.retryTimer.stop();
		} catch (Exception e) {
		}
	}
}